use serde::{Deserialize, Serialize};
use tokio::{
    fs::File,
    io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt, BufReader},
    sync::RwLock,
};
use tonic::{Request, Response, Status};
use tonic_inf::api::{
    byte_buffer_msg::{self, Bfmsg},
    download_response::*,
    send_content_msg::{self},
    tunnel_service_server::*,
    BrowserRequest, BrowserResponse, ByteBufferMsg, ByteBufferStop, DownloadRequest,
    DownloadResponse, FileCheck, FileMeta, RangeDownloadRequest, ResponseBuffer, ResponseStart,
    SendContentMsg, SendContentStop,
};

use path_clean::PathClean;
use std::{
    collections::HashMap, convert::TryFrom, convert::TryInto, path::PathBuf, pin::Pin,
    time::SystemTime, u64, vec,
};

use crate::imp::coolink_sdk_httpd_imp::get_http_host_inner;
use crate::imp::coolink_sdk_httpd_imp::get_http_srv_root_dir;
use crate::imp::coolink_sdk_httpd_imp::SYMBOL_PREFIX;
use crate::imp::tonicrpc::tonicserver::{
    get_or_set_bytebuffer_callback, get_or_set_bytebuffer_sender, get_or_set_content_sender,
    get_or_set_remote_content_callback,
};
use crate::imp::tonicrpc::tonictunnel::{
    select_bytebuf_tunnel_duplex, select_content_tunnel_duplex,
};
use crate::utils::ids;
use crate::utils::symbol_link_util;

use crate::utils::file_util::{file_match_sha256, FILE_BUFFER_SIZE};
use anyhow::Result;
use async_stream::try_stream;
use futures_io::SeekFrom;
use futures_util::{future::Either, pin_mut, StreamExt};

type Stream<T> =
    Pin<Box<dyn futures_core::Stream<Item = std::result::Result<T, Status>> + Send + 'static>>;
// to download file meta information
#[derive(Debug, Deserialize, Serialize)]
pub struct FileMetaInfo {
    // file name
    pub file_name: String,
    // file size
    pub file_size: u64,
    // file create time
    pub creation_time: u64,
    // file last modification time
    pub last_modification_time: u64,
    // file md5
    pub file_sha256: Vec<u8>,
    // file type
    pub mime: String,
    // file or dir
    pub is_file: bool,
}

// from file meta info to metainfo used for tonic
impl From<FileMetaInfo> for FileMeta {
    fn from(info: FileMetaInfo) -> Self {
        let FileMetaInfo {
            file_size,
            file_name,
            creation_time,
            last_modification_time,
            file_sha256,
            mime,
            is_file,
        } = info;
        FileMeta {
            file_name,
            file_size,
            creation_time,
            last_modification_time,
            mime,
            file_sha256,
            is_file,
        }
    }
}

lazy_static! {
    // file download base path,means www root dir.
    static ref DOWNLOAD_BASE_PATH: RwLock<String> = RwLock::new("./".to_owned());

    // add a download path to map that can be send to remote.
    static ref DOWNLOAD_PATH_MAP: std::sync::RwLock<HashMap<String, String>> = std::sync::RwLock::new(HashMap::new());
}

// set download path,default to ./ path,
pub async fn set_download_base_path(path: String) {
    *DOWNLOAD_BASE_PATH.write().await = path;
}

// file path add/remove for map
#[derive(Copy, Clone)]
pub enum MapOperate {
    Add = 1,
    Delete,
}

// form u8 to MapOperate
impl TryFrom<u8> for MapOperate {
    type Error = anyhow::Error;

    fn try_from(value: u8) -> Result<Self, Self::Error> {
        match value {
            x if x == MapOperate::Add as u8 => Ok(MapOperate::Add),
            x if x == MapOperate::Delete as u8 => Ok(MapOperate::Delete),
            _ => Err(anyhow::anyhow!("invild operate")),
        }
    }
}

pub fn reset_download_path_map() -> Result<(), anyhow::Error> {
    let mut map = DOWNLOAD_PATH_MAP
        .try_write()
        .map_err(|e| anyhow::anyhow!(e.to_string()))?;
    map.clear();

    Ok(())
}

// add or remove to a path to download map,and return unique file id.
pub fn operate_download_path_map(operate: u8, path_str: String) -> Result<String, anyhow::Error> {
    let operate = operate.try_into()?;
    let mut map = DOWNLOAD_PATH_MAP
        .try_write()
        .map_err(|e| anyhow::anyhow!(e.to_string()))?;
    let id = match operate {
        MapOperate::Add => {
            let mut found = None;
            for (key, value) in map.iter() {
                if value == &path_str {
                    found = Some(key);
                    break;
                }
            }
            match found {
                Some(id) => id.clone(),
                None => {
                    let path = PathBuf::from(path_str.clone());
                    anyhow::ensure!(path.exists(), "add path not exists");
                    anyhow::ensure!(path.is_file(), "add path not file");
                    // create a symbol link in root path
                    let id = if let Some(root_dir) = get_http_srv_root_dir() {
                        let symbol_id = if let Ok(symbol_file_name) =
                            symbol_link_util::custom_file_symlink(&path_str, &root_dir)
                        {
                            let symbol_path_url_encoder = urlencoding::encode(&symbol_file_name);

                            let host_with_symbol_path_opt = if let Ok(host) = get_http_host_inner()
                            {
                                let symbol_path_host_addr =
                                    format!("{}?symlink={}", host, symbol_path_url_encoder);
                                log::info!(
                                    "operate_download_path_map symbol_path_host_addr = {:?}",
                                    symbol_path_host_addr
                                );
                                Some(symbol_path_host_addr)
                            } else {
                                None
                            };

                            if host_with_symbol_path_opt.is_some() {
                                let host_with_symbol_path = host_with_symbol_path_opt.unwrap();
                                let id_encorder =
                                    urlencoding::encode(&host_with_symbol_path).to_string();
                                let file_id = SYMBOL_PREFIX.to_owned() + &id_encorder;
                                file_id
                            } else {
                                ids::id_generate_xbit(6)
                            }
                        } else {
                            ids::id_generate_xbit(6)
                        };
                        symbol_id
                    } else {
                        ids::id_generate_xbit(6)
                    };

                    map.insert(id.clone(), path_str);
                    id
                }
            }
        }
        MapOperate::Delete => map.remove(&path_str).unwrap_or_default(),
    };
    Ok(id)
}

// download file implementation
async fn download_file(
    relative_file_path: Option<String>,
    file_id: Option<String>,
    file_checks: Vec<FileCheck>,
    need_sha256: bool,
    start: u64,
    end: u64,
) -> Result<Response<Stream<DownloadResponse>>, Status> {
    // get the file path ready to download
    let file_path = match (relative_file_path, file_id) {
        (_, Some(file_id)) => {
            let path = DOWNLOAD_PATH_MAP
                .try_read()
                .map_err(|e| Status::unavailable(e.to_string()))?
                .get(&file_id)
                .ok_or(Status::not_found("file not register"))?
                .to_owned();
            let path = PathBuf::from(path);
            if !path.exists() {
                log::warn!("DOWNLOAD_PATH_MAP not exsit, {:?}", path);
                return Err(Status::unavailable("DOWNLOAD_PATH_MAP path not exsit"));
            }
            path
        }
        (Some(relative_file_path), _) => {
            let base_path = DOWNLOAD_BASE_PATH.read().await.clone();
            let mut file_path = PathBuf::from(base_path);
            if !file_path.exists() {
                log::warn!("DOWNLOAD_BASE_PATH not exsit, {:?}", file_path);
                return Err(Status::unavailable("DOWNLOAD_BASE_PATH not exsit"));
            }
            file_path.push(relative_file_path.trim_start_matches('/'));
            file_path
        }
        _ => {
            return Err(Status::invalid_argument(
                "relative_file_path and file_id not set",
            ));
        }
    };

    log::info!(
        "send_file file_path:{:?} file_checks:{:?}",
        file_path,
        file_checks
    );
    // if set need calcute sha256 or file check len large than 0,
    // then need calcute the sha256
    let need_sha256 = need_sha256 || file_checks.len() > 0;
    let (
        FileMetaInfo {
            file_size,
            file_name,
            creation_time,
            last_modification_time,
            file_sha256,
            mime,
            ..
        },
        match_res,
        _,
    ) = file_match_sha256(&file_path, file_checks, need_sha256)
        .await
        .map_err(|e| Status::not_found(e.to_string()))?;
    let mut start_download_pos = start;
    let mut file_match_index: i32 = -1;
    // if is match
    for (i, (is_match, pos)) in match_res.into_iter().enumerate() {
        // resume for download.
        if is_match && pos > start_download_pos {
            start_download_pos = pos;
            file_match_index = i as i32;
        }
    }
    let total_bytes = if end == 0 {
        file_size
    } else if end >= start_download_pos {
        end
    } else {
        file_size
    };
    let stream = try_stream! {
        let mut bytes_left = (total_bytes - start_download_pos) as usize;
        let response_start = ResponseStart {
            mime,
            file_name,
            file_size,
            start_download_pos,
            file_match_index,
            file_sha256,
            creation_time,
            last_modification_time,
        };
        log::info!("send_file response_start: {:?}", &response_start);
        let item: DownloadResponse = DownloadResponse {
            download_result: Some(DownloadResult::Start(response_start)),
        };
        yield item; // return download response

        let file = File::open(file_path).await?;
        log::info!("send_file bytes_left: {}", bytes_left);
        let mut buf = [0u8; FILE_BUFFER_SIZE];

        let mut read_buffer = BufReader::new(file);
        let mut packet_num = 0;
        let start_time = SystemTime::now();
        let mut each_start_time = start_time.clone();
        if (start_download_pos > 0) {
            read_buffer.seek(SeekFrom::Start(start_download_pos)).await?;
        }
        while bytes_left > 0 {
            let buffer_size = if bytes_left > FILE_BUFFER_SIZE { FILE_BUFFER_SIZE } else { bytes_left };
            let size = read_buffer.read(&mut buf[..buffer_size]).await?;
            // let now = SystemTime::now();
            // let read_time_us = now.duration_since(each_start_time).unwrap_or_default().as_micros();
            // each_start_time = now;
            log::trace!("send_file read size: {}", size);
            if size == 0 {
                log::trace!("send_file size == 0");
                break;
            }
            bytes_left = bytes_left - size;
            packet_num = packet_num + 1;
            let item = DownloadResponse {
                download_result: Some(DownloadResult::Buffer(ResponseBuffer {
                    buffer: buf[..size].to_vec(),
                    is_finish: bytes_left == 0
                })),
            };
            yield item; // return download buffer
            let now = SystemTime::now();
            let send_time_us = now
                .duration_since(each_start_time)
                .unwrap_or_default()
                .as_micros();
            each_start_time = now;
            log::trace!("send_file send buffer send_time_us:{}", send_time_us);
        }
        // log send file time milliseconds.
        let mut all_send_time_ms = SystemTime::now().duration_since(start_time).unwrap_or_default().as_millis();
        if all_send_time_ms == 0 {
            all_send_time_ms = 1
        }
        log::warn!(
            "send_file all_send_time_ms: {} speed: {}KB/S packet_num: {} ",
            all_send_time_ms,
            file_size / all_send_time_ms as u64,
            packet_num
        );
    };
    Ok(Response::new(Box::pin(stream) as Stream<DownloadResponse>))
}

#[derive(Debug)]
pub struct TonicServerImp {}

// implementation of tonic traits,used for tonic client call.
#[tonic::async_trait]
impl TunnelService for TonicServerImp {
    // tonic service implement browser file request
    async fn browser_file(
        &self,
        request: Request<BrowserRequest>,
    ) -> Result<Response<BrowserResponse>, Status> {
        let BrowserRequest {
            relative_path,
            need_sha256,
            goto_folder,
        } = request.into_inner();
        log::info!(
            "browser_file relative_path:{} need_sha256:{} goto_folder:{}",
            relative_path,
            need_sha256,
            goto_folder
        );
        let base_path = DOWNLOAD_BASE_PATH.read().await.clone();
        let mut file_path = PathBuf::from(base_path.clone());
        if relative_path != "" || relative_path != "." || relative_path != "/" {
            file_path.push(relative_path.trim_start_matches('/'));
        }

        let mut relative_path = PathBuf::from(relative_path);
        let mut file_list = vec![];
        let mut file = None;

        if goto_folder != "" || goto_folder != "." || goto_folder != "/" {
            file_path.push(goto_folder.trim_start_matches('/'));
            relative_path.push(goto_folder.trim_start_matches('/'));
        }
        file_path = file_path.clean();
        relative_path = relative_path.clean();
        if !file_path.starts_with(base_path) {
            return Err(Status::permission_denied("no permission over base path"));
        }

        if file_path.is_dir() {
            let mut list = tokio::fs::read_dir(file_path).await?;
            while let Some(entry) = list.next_entry().await? {
                let (info, _, _) = file_match_sha256(&entry.path(), vec![], need_sha256)
                    .await
                    .map_err(|e| Status::not_found(e.to_string()))?;
                file_list.push(info.into());
            }
        } else {
            let (info, _, _) = file_match_sha256(&file_path, vec![], need_sha256)
                .await
                .map_err(|e| Status::not_found(e.to_string()))?;
            file = Some(info.into());
        }

        Ok(Response::new(BrowserResponse {
            relative_file_path: relative_path.to_str().unwrap_or_default().to_owned(),
            file_list,
            file,
        }))
    }

    type range_downloadStream = Stream<DownloadResponse>;
    // tonic service implement range download file request
    async fn range_download(
        &self,
        request: Request<RangeDownloadRequest>,
    ) -> Result<Response<Self::downloadStream>, Status> {
        let RangeDownloadRequest {
            relative_file_path,
            need_sha256,
            start,
            end,
            file_id,
        } = request.into_inner();
        log::info!(
            "range_download relative_file_path:{:?} start:{} end:{}",
            relative_file_path,
            start,
            end
        );
        download_file(relative_file_path, file_id, vec![], need_sha256, start, end).await
    }

    type downloadStream = Stream<DownloadResponse>;
    // tonic service implement download file request
    async fn download(
        &self,
        request: Request<DownloadRequest>,
    ) -> Result<Response<Self::downloadStream>, Status> {
        // let (mut tx, rx) = mpsc::channel(16);
        let DownloadRequest {
            relative_file_path,
            file_checks,
            file_id,
        } = request.into_inner();
        log::info!(
            "download relative_file_path:{:?} file_checks:{:?}",
            relative_file_path,
            file_checks
        );
        download_file(relative_file_path, file_id, file_checks, true, 0, 0).await
        // Ok(Response::new(ReceiverStream::new(rx)))
    }

    type send_contentStream = Stream<SendContentMsg>;
    // tonic service implement send content request
    async fn send_content(
        &self,
        request: Request<tonic::Streaming<SendContentMsg>>,
    ) -> Result<Response<Self::send_contentStream>, Status> {
        log::info!("send_content server start ");
        let stream = try_stream! {
            let (outbound_remote_tx, mut outbound_remote_rx) = tokio::sync::mpsc::unbounded_channel::<SendContentMsg>();

            let res = tokio::spawn(async {
                let mut inbound_remote = request.into_inner();
                let (inbound_remote_tx, inbound_remote_rx) = tokio::sync::mpsc::unbounded_channel();
                let start_msg = inbound_remote.message().await?.ok_or(anyhow::anyhow!("send_content start with none msg"))?;
                let SendContentMsg {msg} = &start_msg;
                let msg = msg.clone().ok_or(anyhow::anyhow!("send_content start msg none"))?;
                // received from remote client,received SendContentMsg
                let tunnel_id = match msg {
                    // if msg = start.
                    send_content_msg::Msg::Start(start) => {
                        start.tunnel_id
                    },
                    _ => {
                        // "".to_owned()
                        anyhow::bail!("send_content start not with start msg");
                    }
                };
                inbound_remote_tx.send(start_msg).map_err(|e| Status::unknown(e.to_string()))?;
                let inbound_remote_handler = async {
                    while let Some(msg) = inbound_remote.message().await? {
                        inbound_remote_tx.send(msg)?;
                    }
                    let res: Result<(), anyhow::Error> = Ok(());
                    res
                };

                let outbound_user_callback_tx = get_or_set_remote_content_callback(false, None)
                .await
                .ok_or(anyhow::anyhow!(
                    "outbound_user_tx not set, maybe service stop?",
                ))?;

                let start_content_exchange_handler = select_content_tunnel_duplex(None, inbound_remote_rx, outbound_remote_tx, outbound_user_callback_tx.clone());
                pin_mut!(inbound_remote_handler, start_content_exchange_handler);

                // wait for selet finish.
                let res = match futures_util::future::select(inbound_remote_handler, start_content_exchange_handler).await {
                    Either::Left((value1, _)) => value1,
                    Either::Right((value2, _)) => value2,
                };

                // stop content sender.
                get_or_set_content_sender(true, &tunnel_id, None)?;
                outbound_user_callback_tx.send(send_content_msg::Msg::Stop(SendContentStop{tunnel_id, is_confirm: true}))?;
                res
            });

            // streamming...
            while let Some(item) = outbound_remote_rx.recv().await {
                log::debug!("outbound_remote_rx.recv() item: {:?}", item);
                yield item;
            }

            res.await
                .map_err(|e| Status::unknown(e.to_string()))?.map_err(|e| Status::unknown(e.to_string()))?;
        };
        Ok(Response::new(Box::pin(stream) as Self::send_contentStream))
    }

    type send_bytebufferStream = Stream<ByteBufferMsg>;
    // tonic service implement send content request
    async fn send_bytebuffer(
        &self,
        request: Request<tonic::Streaming<ByteBufferMsg>>,
    ) -> Result<Response<Self::send_bytebufferStream>, Status> {
        log::info!("send_bytebuffer server start ");
        let stream = try_stream! {
            let (outbound_remote_tx, mut outbound_remote_rx) = tokio::sync::mpsc::unbounded_channel::<ByteBufferMsg>();

            let res = tokio::spawn(async {
                let mut inbound_remote = request.into_inner();
                let (inbound_remote_tx, inbound_remote_rx) = tokio::sync::mpsc::unbounded_channel();
                let start_msg = inbound_remote.message().await?.ok_or(anyhow::anyhow!("send_bytebuffer start with none msg"))?;
                let ByteBufferMsg {bfmsg} = &start_msg;
                let msg = bfmsg.clone().ok_or(anyhow::anyhow!("send_bytebuffer start msg none"))?;
                // received from remote client,received ByteBufferMsg
                let tunnel_id = match msg {
                    // if msg = start.
                    Bfmsg::BfStart(start) => {
                        start.tunnel_id
                    },
                    _ => {
                        // "".to_owned()
                        anyhow::bail!("send_bytebuffer start not with start msg");
                    }
                };
                inbound_remote_tx.send(start_msg).map_err(|e| Status::unknown(e.to_string()))?;
                let inbound_remote_handler = async {
                    while let Some(msg) = inbound_remote.message().await? {
                        inbound_remote_tx.send(msg)?;
                    }
                    let res: Result<(), anyhow::Error> = Ok(());
                    res
                };

                let outbound_user_callback_tx = get_or_set_bytebuffer_callback(false, None)
                .await
                .ok_or(anyhow::anyhow!(
                    "outbound_user_tx not set, maybe service stop?",
                ))?;

                let start_bytebuffer_exchange_handler = select_bytebuf_tunnel_duplex(None, inbound_remote_rx, outbound_remote_tx, outbound_user_callback_tx.clone());
                pin_mut!(inbound_remote_handler, start_bytebuffer_exchange_handler);

                // wait for selet finish.
                let res = match futures_util::future::select(inbound_remote_handler, start_bytebuffer_exchange_handler).await {
                    Either::Left((value1, _)) => value1,
                    Either::Right((value2, _)) => value2,
                };

                // stop content sender.
                get_or_set_bytebuffer_sender(true, &tunnel_id, None)?;
                outbound_user_callback_tx.send(byte_buffer_msg::Bfmsg::BfStop(ByteBufferStop{tunnel_id, is_confirm: true}))?;
                res
            });

            // streamming...
            while let Some(item) = outbound_remote_rx.recv().await {
                log::debug!("outbound_remote_rx.recv() item: {:?}", item);
                yield item;
            }

            res.await
                .map_err(|e| Status::unknown(e.to_string()))?.map_err(|e| Status::unknown(e.to_string()))?;
        };
        Ok(Response::new(
            Box::pin(stream) as Self::send_bytebufferStream
        ))
    }
}
